package com.iot.mqtt.test;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.io.UnsupportedEncodingException;
import java.util.*;

/**
 * @author Cao.ning
 * @date 2023/5/17
 */
public class HainayunClient {
    private static final Logger logger = LoggerFactory.getLogger(HainayunClient.class);

    @Autowired
    public ThreadPoolTaskExecutor threadPoolTaskExecutor;


    public static final String broker = "tcp://58.58.38.214:1883";
    public static final String username = "hainayun_test";
    public static final String password = "hainayun_test";
    public static final int timeout = 8000;
    public static final int keepAliveInterval = 20;

    public static final String GET_TOPIC = "aiotup/adapter/raw/qpxzns6jamwsrxls/%s/GET2";
    public static final String SET_TOPIC = "aiotup/adapter/raw/qpxzns6jamwsrxls/%s/SET2";

    public static final Map<String, MqttClient> clientMap = new HashMap();
    public static final Map<String, Timer> timerMap = new HashMap();

    /*public void init(List<String> clientIds){
        for(String clientId : clientIds){
            this.initClientConnect(clientId);
        }
    }*/

    public static void initClientConnect(String clientId) {
        try {
            MqttClient mqttClient = new MqttClient(broker, clientId, new MemoryPersistence());
            MqttConnectOptions mqttConnectOptions = new MqttConnectOptions();
            mqttConnectOptions.setUserName(username);
            mqttConnectOptions.setPassword(password.toCharArray());
            mqttConnectOptions.setServerURIs(new String[]{ broker });
            mqttConnectOptions.setConnectionTimeout(timeout);// 设置超时时间
            mqttConnectOptions.setKeepAliveInterval(keepAliveInterval);// 设置会话心跳时间

            mqttClient.setCallback(new MqttCallbackExtended() {


                @Override
                public void connectionLost(Throwable throwable) {
                    try {
                        Thread.sleep(10000);
                        mqttClient.reconnect();
                    } catch (InterruptedException | MqttException e) {
                        e.printStackTrace();
                    }
                }

                @Override
                public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
                    logger.info("海纳云测试{} 收到消息：{}", clientId, new String(mqttMessage.getPayload()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

                }

                @Override
                public void connectComplete(boolean b, String s) {
                    Timer timer = timerMap.getOrDefault(clientId, new Timer());
                    timerMap.put(clientId, timer);
                    timer.schedule(new TimerTask() {
                        @Override
                        public void run() {
                            logger.info("海纳云测试{}发布心跳，mqtt服务器：{}", clientId, mqttClient.isConnected() ? "在线" : "离线");
                            HainayunClient.heartBeat(clientId);
                        }
                    }, 1000, 180000);//1秒后启动，隔5分钟自检一次
                }
            });
            mqttClient.connect(mqttConnectOptions);
            logger.info("<<<<<<<<<< 海纳云测试mqtt客户端已连接 >>>>>>>>>>>> serverUrl: {}， clientId: {} ", broker, clientId);

            clientMap.put(clientId, mqttClient);
        } catch (MqttException e) {
            logger.info("<<<<<<<<<< 初始化海纳云测试mqtt客户端异常 >>>>>>>>>>>> clientId: {} ", clientId);
            e.printStackTrace();
        }
    }

    public static void heartBeat(String clientId){
        MqttClient mqttClient = clientMap.get(clientId);
        if(mqttClient != null && mqttClient.isConnected()){
            try {
                MqttMessage message = new MqttMessage("online".getBytes("UTF-8"));
                message.setQos(2);
                mqttClient.publish(String.format(GET_TOPIC, clientId), message);
            } catch (UnsupportedEncodingException | MqttException e) {
                e.printStackTrace();
            }
        }
    }
}
